unit uSBDUnit_JobMillImpl;
interface
uses Classes, uSBDUnit_SyncObjects, SysUtils, Windows, uSBDUnit_JobMill,
     uSBD_ServiceProvider;

type


TJobMill = class( TThread, IInterface, IJobMill)
  strict private
    FRefCount: integer;
    function QueryInterface(const IID: TGUID; out Obj): HResult; stdcall;
    function _AddRef: Integer; stdcall;
    function _Release: Integer; stdcall;

  private
    FInputQueue : TTransThread_ObjectQueue;
    FOutputQueue: TTransThread_ObjectQueue;
    FPoisonPill : TSBD_Semaphore;
    FhasKilled  : boolean;
    FhasPoisoned: boolean;

    {$REGION 'Property accessors'}
    FOnPollOutputThread: TNotifyEvent;
    procedure SetTimeOut( Value_ms: integer);
    procedure SetOnPollOutputThread( Value: TNotifyEvent);
    {$ENDREGION}
    procedure JobBreathe( var doAbortExecute: boolean);

  protected
    procedure Execute; override;

  public
    FWindowHndl : HWND;
    FMessageNum : Cardinal;

    [Configuration] constructor ServiceModeCreate;
    constructor Create( TimeOut_ms1: integer; OnPollOutputThread1: TNotifyEvent);
    destructor  Destroy; override;

    procedure QueueJob( Job: TJob);
    function  WaitPop( TimeOut: integer = UsePropertyDefault): boolean;   // Return True if got a returned job.
    function  PeekPop: boolean;   // Return True if got a returned job.
    procedure KillThread;
    procedure Poison;

    property  TimeOut_ms: integer    write SetTimeOut;
    property  OnPollOutputThread: TNotifyEvent read FOnPollOutputThread write FOnPollOutputThread;
  end;


procedure RegisterServices( const Registrar: IServiceProvider);
// ^-- Registers (IJobMill,'')


implementation







procedure RegisterServices( const Registrar: IServiceProvider);
begin
Registrar.RegisterServiceClass( TJobMill, IJobMill).Free;
end;

{ TJobMill }
constructor TJobMill.ServiceModeCreate;
begin
// Typically, FRefCount = 2 at this point.
Create( 2000, nil)
end;


constructor TJobMill.Create( TimeOut_ms1: integer; OnPollOutputThread1: TNotifyEvent);
begin
// Do not set FRefCount.
FreeOnTerminate := False;  // Client is responsible for destruction.
FPoisonPill  := TSBD_Semaphore.CreateSimple;
FInputQueue  := TTransThread_ObjectQueue.Create( FPoisonPill);
FOutputQueue := TTransThread_ObjectQueue.Create( nil);
FOutputQueue.TimeOut_ms := -1;
FhasKilled   := False;
FhasPoisoned := False;
TimeOut_ms := TimeOut_ms1;
OnPollOutputThread := OnPollOutputThread1;
inherited Create( False)
end;


procedure TJobMill.KillThread;
var
  dtEndPatience: TDateTime;
  j: Integer;
  ExceptionOccured: boolean;
begin
if FhasKilled then exit;
FhasKilled := True;
Poison;  // Releases the poison semaphore and calls Terminate.
while Suspended do
  begin
  Resume;
  Sleep(1);
  end;
dtEndPatience := Now + (0.5 / SecsPerDay);
ExceptionOccured := False;
while ((FOutputQueue.Count > 0) or (not Finished)) and (Now <= dtEndPatience) do
  for j := 1 to 1000 do
    begin
    Sleep(1);
    if (not ExceptionOccured) and assigned( FOnPollOutputThread) then
      try
        FOnPollOutputThread( self)
      except
        ExceptionOccured := True
      end;
    if (FOutputQueue.Count = 0) and Finished then break
    end;
if FOutputQueue.Count > 0 then
  FOutputQueue.BrutalClose;
if Finished and (GetCurrentThreadID = MainThreadID) then
  CheckSynchronize( 100);
if not Finished then
  TerminateThread( ThreadId, 1);
FreeAndNil( FInputQueue);
FreeAndNil( FOutputQueue);
FreeAndNil( FPoisonPill)
end;

destructor TJobMill.Destroy;
// Warning: Do not destroy the Job Mill until the output queue has
//  been naturally depleted.
var
  Count: integer;
begin
KillThread;
FreeAndNil( FInputQueue);
FreeAndNil( FOutputQueue);
FreeAndNil( FPoisonPill);
inherited
end;


procedure TJobMill.Execute;
var
  Obj: TObject;
  Job: TJob;
  isThisPoisoned: boolean;
begin
Obj := nil;
while (not Terminated) and (not FInputQueue.isClosed) do
  begin
  isThisPoisoned := False;
  Obj := FInputQueue.WaitPop;
  if Obj is TJob then
    begin
    try
      Job := TJob( Obj);
      Job.FOnBreatheProc := JobBreathe;
      Job.Execute_InThread
    except
      on EAbortJob do ;
      on e: Exception do
        Job.CaptureException( e, ExceptAddr)
      end;
    Job.FOnBreatheProc := nil;
    isThisPoisoned := joPoisoned in Job.FOptions;
    if joNoReturn in Job.FOptions then
        try
          FreeAndNil( Obj)
        except end
      else
        begin
        FOutputQueue.Push( Obj);
        if (FWindowHndl <> 0) and (FMessageNum <> 0) then
          PostMessage( FWindowHndl, FMessageNum, 0, 0);
        Obj := nil
        end
    end;
  if isThisPoisoned then
    Poison
  end;
Obj.Free
end;



procedure TJobMill.JobBreathe( var doAbortExecute: boolean);
begin
doAbortExecute := Terminated or FInputQueue.isClosed or
                  FPoisonPill.isSignalled;
if doAbortExecute then
  raise EAbortJob.Create('Aborting job');
end;


procedure TJobMill.Poison;
begin
if FhasPoisoned then exit;
FhasPoisoned := True;
FPoisonPill.Release;
Terminate
end;


function TJobMill.QueryInterface( const IID: TGUID; out Obj): HResult;
begin
if GetInterface( IID, Obj) then
    result := 0
  else
    result := E_NOINTERFACE
end;

procedure TJobMill.QueueJob( Job: TJob);
begin
if not assigned( Job) then exit;
if assigned( FInputQueue) and (not FInputQueue.isClosed) then
    FInputQueue.Push( Job)
  else
    raise Exception.Create('JobMill is closed.');
end;


procedure TJobMill.SetOnPollOutputThread( Value: TNotifyEvent);
begin
FOnPollOutputThread := Value
end;

procedure TJobMill.SetTimeOut( Value_ms: integer);
begin
FOutputQueue.TimeOut_ms := Value_ms
end;

function TJobMill.WaitPop( TimeOut: integer = UsePropertyDefault): boolean;
var
  Obj: TObject;
begin
if assigned( FOutputQueue) then
    Obj := FOutputQueue.WaitPop( TimeOut)
  else
    Obj := nil;
result := Obj is TJob;
if result then
  TJob( Obj).Execute_OutOfThread;
Obj.Free
end;





function TJobMill._AddRef: Integer;
begin
result := InterlockedIncrement( FRefCount)
end;

function TJobMill._Release: Integer;
begin
result := InterlockedDecrement( FRefCount);
if result = 0 then
  Destroy
end;

function TJobMill.PeekPop: boolean;
var
  Obj: TObject;
begin
if assigned( FOutputQueue) then
    Obj := FOutputQueue.PeekPop
  else
    Obj := nil;
result := Obj is TJob;
if result then
  TJob( Obj).Execute_OutOfThread;
Obj.Free
end;


end.
